package com.haiyou.data.common.accessor.redis.redisson.bucket;

import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.redisson.api.RScoredSortedSet;
import org.redisson.client.protocol.ScoredEntry;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.util.reflect.AnnotationFinder;
import com.haiyou.data.common.accessor.annotation.SizePolicy;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.persister.mysql.help.DatabaseHelper;
import com.haiyou.data.common.persister.mysql.help.RollTableQueryHelper;
import com.haiyou.data.common.persister.mysql.annotation.Table.RollPolicy;


/**
 * 分布式对象桶+分布式倒序计分容器，适合用在经常需要调用getAll函数，元素缓存存活时间较短的排行榜
 * 
 * @author Administrator
 *
 * @param <K>
 * @param <V>
 */
public abstract class ScoredBucketAccessor<V extends Entity<String>> extends RedissonBucketAccessor<V> {

	private static final String SCOREDSORTED_SUFFIX = "ScoredSortedView";

	protected RScoredSortedSet<String> scoredSortedSet;// 排行榜的值只存key
	protected int greed_length;

	@Override
	public void init() {
		super.init();

		SizePolicy sizePolicy = AnnotationFinder.findAnnotation(getClass(), SizePolicy.class);
		if (sizePolicy == null) {
			greed_length = 1000;
		} else {
			greed_length = sizePolicy.maximumSize();
		}
		greed_length = Math.max(greed_length, 1);
		greed_length = Math.min(greed_length, 100000);

		scoredSortedSet = redissonClient
				.getScoredSortedSet(RKeys.createRKey(RScoredSortedSet.class, keyPrefix, SCOREDSORTED_SUFFIX));
	}

	/**
	 * 计算分数
	 * 
	 * @param vlaue
	 * @return
	 */
	protected abstract double calcScore(V vlaue);

	/**
	 * 名次
	 * 
	 * @param key
	 * @return
	 */
	public int rank(String key) {
		return scoredSortedSet.rank(key);
	}

	/**
	 * 分数
	 * 
	 * @param key
	 * @return
	 */
	public double score(String key) {
		return scoredSortedSet.getScore(key);
	}

	/**
	 * 返回 score1 和 score2 之间的所有条目
	 * 
	 * @param score1
	 * @param score2
	 * @return
	 */
	public List<V> subList(double score1, double score2) {
		score1 = Math.min(score1, score2);
		score2 = Math.max(score1, score2);
		Collection<ScoredEntry<String>> collection = scoredSortedSet.entryRangeReversed(score1, true, score2, true);
		List<String> result = Lists.newArrayListWithCapacity(collection.size());
		Iterator<ScoredEntry<String>> iterator = collection.iterator();
		while (iterator.hasNext()) {
			result.add(iterator.next().getValue());
		}
		return getAll(result).values().stream().collect(Collectors.toList());
	}

	/**
	 * 返回 index1 和 index2 之间的所有条目
	 * 
	 * @param index1
	 * @param index2
	 * @return
	 */
	public List<V> subList(int index1, int index2) {
		index1 = Math.min(index1, index2);
		index2 = Math.max(index1, index2);
		if (index1 == index2)
			return Collections.emptyList();
		if (index2 <= scoredSortedSet.size()) {
			return getSubListFromRScoredSortedSet(index1, Math.min(index2, scoredSortedSet.size()));
		}
		if (index1 >= scoredSortedSet.size()) {
			return getSubListFromMySql(index1, index2);
		}
		if (index1 < scoredSortedSet.size() && index2 > scoredSortedSet.size()) {
			List<V> result = Lists.newArrayListWithCapacity(index2 - index1);
			List<V> redisList = getSubListFromRScoredSortedSet(index1, scoredSortedSet.size());
			List<V> mySqlList = getSubListFromMySql(scoredSortedSet.size(), index2);
			result.addAll(redisList);
			result.addAll(mySqlList);
			return result;
		}
		return Collections.emptyList();
	}

	private List<V> getSubListFromRScoredSortedSet(int index1, int index2) {
		Collection<ScoredEntry<String>> collection = scoredSortedSet.entryRangeReversed(index1, index2);
		List<String> result = Lists.newArrayListWithCapacity(collection.size());
		Iterator<ScoredEntry<String>> iterator = collection.iterator();
		while (iterator.hasNext())
			result.add(iterator.next().getValue());
		List<V> resultList = getAll(result).values().stream().collect(Collectors.toList());
		return resultList;
	}

	private List<V> getSubListFromMySql(int index1, int index2) {
		List<V> result = null;
		try {
			if (persister.getTableMeta().getRollPolicy() != RollPolicy.NONE) {
				return result = RollTableQueryHelper.queryByLimits(valueClass, index1, index2, true);
			} else {
				return result = persister.queryByLimits(DatabaseHelper.getOnlyReallyName(persister.getDatabaseMeta()),
						persister.getTableMeta().getSimpleName(), index1, index2, true);
			}
		} finally {
			if (result != null && !result.isEmpty()) {
				for (int i = 0; i < result.size(); i++) {
					if (scoredSortedSet.size() < greed_length)
						scoredSortedSet.add(calcScore(result.get(i)), result.get(i).getId());
				}
			}
		}
	}

	private void reSize() {
		while (scoredSortedSet.size() > greed_length) {
			scoredSortedSet.pollFirstAsync();
		}
	}

	@Override
	public V put(String key, V value) {
		scoredSortedSet.addAsync(calcScore(value), key);
		reSize();
		return super.put(key, value);
	}

	@Override
	public void putAll(Map<? extends String, ? extends V> m) {
		Map<String, Double> scoredMap = Maps.newHashMap();
		m.forEach((k, v) -> scoredMap.put(k, calcScore(v)));
		scoredSortedSet.addAll(scoredMap);
		reSize();
		super.putAll(m);
	}

	@Override
	public V remove(Object key) {
		scoredSortedSet.remove(key);
		return super.remove(key);
	}

	@Override
	public void removeAll(Iterable<? extends String> keys) {
		scoredSortedSet.removeAll(Sets.newHashSet(keys));
		super.removeAll(keys);
	}

}
